package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.util.NameUnusualUtil;
import org.apache.flink.table.functions.AggregateFunction;

import java.util.Iterator;

public class UserknNameUnusualNum extends AggregateFunction<Integer, NameUnusualUtil> {
    @Override
    public NameUnusualUtil createAccumulator() {
        return new NameUnusualUtil();
    }

    @Override
    public Integer getValue(NameUnusualUtil nameUnusualUtil) {
        return nameUnusualUtil.getCount();
    }

    public void accumulate(NameUnusualUtil nameUnusualUtil, String name) {
        if (!nameUnusualUtil.exitsName(name)) {
            //异常姓名数加1
            if (isFigureName(name) || name.length() > 5 || isEnglishName(name) || isSpecialName(name)) {
                nameUnusualUtil.addName(name);
                nameUnusualUtil.setCount(nameUnusualUtil.getCount() + 1);
            }
        }
    }

    public boolean isEnglishName(String name) {

        boolean isWord = name.matches("[a-zA-Z]+");
        return isWord;
    }

    public boolean isFigureName(String name) {
        boolean isFigure = name.matches("^[-\\+]?[\\d]*$");
        return isFigure;
    }

    public boolean isSpecialName(String name) {
        boolean isSpecial = !name.matches(".*[0-9a-zA-Z].*|.*[\\u4e00-\\u9fa5].*");
        return isSpecial;
    }

    public void merge(NameUnusualUtil nameUnusualUtil, Iterable<NameUnusualUtil> it) {
        Iterator<NameUnusualUtil> iter = it.iterator();
        while (iter.hasNext()) {
            NameUnusualUtil a = iter.next();
            nameUnusualUtil.setCount(a.getCount()+nameUnusualUtil.getCount());
            nameUnusualUtil.getRepetition().addAll(a.getRepetition());
        }
    }
}